package com.wis.bolt.risk;

import com.wis.TopologyDrive;
import org.apache.commons.lang.StringUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * @Description: traffic_wis
 * @author: fan
 * @Date: Created in 2018/12/17 14:02
 * @Modified By:
 */
public class RISKJiHeDanJiHeBolt extends BaseRichBolt {
    private OutputCollector collector;
    private static String typeProduce;

    @Override
    public void prepare(Map map, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        typeProduce = map.get(TopologyDrive.TYPE_PRODUCE).toString();
    }

    @Override
    public void execute(Tuple tuple) {
        String key = tuple.getStringByField("key");
        String value = tuple.getStringByField("yaosu");
        String id = tuple.getStringByField("id");
        String xt = tuple.getStringByField("xt");
        try {
            if (value.equals("jies")) {
                if (typeProduce.equals("1")) {
                    this.collector.emit(new Values(key.replace("sc", "00"), "over"));
                } else if (typeProduce.equals("2")) {
                    this.collector.emit(new Values(key.replace("sc", "40"), "over"));
                } else if (typeProduce.equals("3")) {
                    this.collector.emit(new Values(key.replace("sc", "80"), "over"));
                }
            } else if (StringUtils.isNotBlank(value) /*&& value.length() > 4*/) {
                String[] ids = id.split("=", -1);
                String[] xts = xt.split("=", -1);
                String[] hang = value.split("=", -1);
                String[][] arrs = new String[hang.length][hang[0].split(",", -1).length];

                for (int i = 0; i < hang.length; i++) {
                    arrs[i] = hang[i].split(",", -1);
                }
                // 假如是预报的话，这儿需要变化
                for (int i = 0; i < arrs[0].length; i++) {// 有多少列
                    List list = new LinkedList();
                    // System.out.println(key + "++++" + value + "+++++" + id + "+++++" + xt);
                    for (int j = 0; j < arrs.length; j++) {
                        if (StringUtils.isNotBlank(arrs[j][i]) && Integer.parseInt(arrs[j][i]) > 0) {
                            try {
                                // System.out.print("--" + xts[j].split(",", -1)[i]);
                                list.add(arrs[j][i] + "_" + xts[j].split(",", -1)[i] + "_" + ids[j]);
                            } catch (Exception e) {
                                System.out.println(arrs[j][i]);
                                e.printStackTrace();
                                System.exit(1);
                            }
                        }
                    }
                    if (list.size() > 0)
                        if (typeProduce.equals("1")) {
                            this.collector.emit(new Values(key.replace("sc", "00"), StringUtils.join(list, ",")));
                        } else if (typeProduce.equals("2")) {
                            this.collector.emit(new Values(key.replace("sc", (i + 1) + "0"), StringUtils.join(list, ",")));
                        } else if (typeProduce.equals("3")) {
                            if (i < 9) {
                                this.collector.emit(new Values(key.replace("sc", "0" + (i + 1)), StringUtils.join(list, ",")));
                            } else {
                                this.collector.emit(new Values(key.replace("sc", (i + 1) + ""), StringUtils.join(list, ",")));
                            }
                        }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            this.collector.ack(tuple);
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("key", "value"));
    }
}